﻿using DotNetCore.CAP;
using DotNetCore.CAP.Internal;
using DotNetCore.CAP.Messages;
using DotNetCore.CAP.Persistence;
using DotNetCore.CAP.Transport;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;

namespace Common.MQ
{
    public class CapCustomDispatcher : IDispatcher
    {
        private readonly IMessageSender sender;
        private readonly CapOptions options;
        private readonly ISubscribeDispatcher executor;
        private readonly ILogger<CapCustomDispatcher> logger;
        private readonly IConfiguration configuration;
        private string[] excuteGroups;
        private string[] pulishTopics;
        private readonly CancellationTokenSource cts = new CancellationTokenSource();

        private Channel<MediumMessage> publishedChannel;
        private Channel<MediumMessage> publishedChannel2;
        private Channel<(MediumMessage, ConsumerExecutorDescriptor)> receivedChannel;
        private Channel<(MediumMessage, ConsumerExecutorDescriptor)> receivedChannel2;

        public CapCustomDispatcher(
            ILogger<CapCustomDispatcher> logger,
            IMessageSender sender,
            IConfiguration config,
            IOptions<CapOptions> options,
            ISubscribeDispatcher executor)
        {
            this.logger = logger;
            this.sender = sender;
            this.options = options.Value;
            this.executor = executor;
            this.configuration = config;
        }

        public void Start(CancellationToken stoppingToken)
        {
            var eg = configuration["MQ:ExtraReceiveGroup"];
            var expp = configuration["MQ:ExtraPulishTopic"];
            excuteGroups = !string.IsNullOrEmpty(eg) ? eg.Split(new char[] { ',' }, StringSplitOptions.RemoveEmptyEntries) : null;
            pulishTopics = !string.IsNullOrEmpty(expp) ? expp.Split(new char[] { ',' }, StringSplitOptions.RemoveEmptyEntries) : null;
            stoppingToken.ThrowIfCancellationRequested();
            stoppingToken.Register(() => cts.Cancel());

            var capacity = options.ProducerThreadCount * 500;
            publishedChannel = Channel.CreateBounded<MediumMessage>(new BoundedChannelOptions(capacity > 5000 ? 5000 : capacity)
            {
                AllowSynchronousContinuations = true,
                SingleReader = options.ProducerThreadCount == 1,
                SingleWriter = true,
                FullMode = BoundedChannelFullMode.Wait
            });

            if (pulishTopics != null && pulishTopics.Any())
            {
                publishedChannel2 = Channel.CreateBounded<MediumMessage>(new BoundedChannelOptions(capacity > 5000 ? 5000 : capacity)
                {
                    AllowSynchronousContinuations = true,
                    SingleReader = options.ProducerThreadCount == 1,
                    SingleWriter = true,
                    FullMode = BoundedChannelFullMode.Wait
                });
            }

            capacity = options.ConsumerThreadCount * 300;
            receivedChannel = Channel.CreateBounded<(MediumMessage, ConsumerExecutorDescriptor)>(new BoundedChannelOptions(capacity > 3000 ? 3000 : capacity)
            {
                AllowSynchronousContinuations = true,
                SingleReader = options.ConsumerThreadCount == 1,
                SingleWriter = true,
                FullMode = BoundedChannelFullMode.Wait
            });

            if (excuteGroups != null && excuteGroups.Any())
            {
                receivedChannel2 = Channel.CreateBounded<(MediumMessage, ConsumerExecutorDescriptor)>(new BoundedChannelOptions(capacity > 3000 ? 3000 : capacity)
                {
                    AllowSynchronousContinuations = true,
                    SingleReader = options.ConsumerThreadCount == 1,
                    SingleWriter = true,
                    FullMode = BoundedChannelFullMode.Wait
                });
            }

            _ = Task.WhenAll(Enumerable.Range(0, options.ProducerThreadCount)
            .Select(_ => Task.Factory.StartNew(() => Sending(stoppingToken), stoppingToken, TaskCreationOptions.LongRunning, TaskScheduler.Default)).ToArray());
            if (pulishTopics != null && pulishTopics.Any())
            {
                _ = Task.WhenAll(Enumerable.Range(0, options.ProducerThreadCount)
                .Select(_ => Task.Factory.StartNew(() => Sending2(stoppingToken), stoppingToken, TaskCreationOptions.LongRunning, TaskScheduler.Default)).ToArray());
            }

            _ = Task.WhenAll(Enumerable.Range(0, options.ConsumerThreadCount)
                .Select(_ => Task.Factory.StartNew(() => Processing(stoppingToken), stoppingToken, TaskCreationOptions.LongRunning, TaskScheduler.Default)).ToArray());

            if (excuteGroups != null && excuteGroups.Any())
            {
                _ = Task.WhenAll(Enumerable.Range(0, options.ConsumerThreadCount)
                .Select(_ => Task.Factory.StartNew(() => Processing2(stoppingToken), stoppingToken, TaskCreationOptions.LongRunning, TaskScheduler.Default)).ToArray());
            }
        }

        public void EnqueueToPublish(MediumMessage message)
        {
            var pp = message.Origin.GetName();
            if (!string.IsNullOrEmpty(pp) && pulishTopics != null && pulishTopics.Any(t => string.Compare(t, pp, true) == 0))
            {
                try
                {
                    if (!publishedChannel2.Writer.TryWrite(message))
                    {
                        while (publishedChannel2.Writer.WaitToWriteAsync(cts.Token).AsTask().ConfigureAwait(false).GetAwaiter().GetResult())
                        {
                            if (publishedChannel2.Writer.TryWrite(message))
                            {
                                return;
                            }
                        }
                    }
                }
                catch (OperationCanceledException)
                {
                    ////Ignore
                }

                return;
            }

            try
            {
                if (!publishedChannel.Writer.TryWrite(message))
                {
                    while (publishedChannel.Writer.WaitToWriteAsync(cts.Token).AsTask().ConfigureAwait(false).GetAwaiter().GetResult())
                    {
                        if (publishedChannel.Writer.TryWrite(message))
                        {
                            return;
                        }
                    }
                }
            }
            catch (OperationCanceledException)
            {
                ////Ignore
            }
        }

        public void EnqueueToExecute(MediumMessage message, ConsumerExecutorDescriptor descriptor)
        {
            var pp = message.Origin.GetGroup();
            if (!string.IsNullOrEmpty(pp) && excuteGroups != null && excuteGroups.Any(t => pp.StartsWith(t, StringComparison.OrdinalIgnoreCase)))
            {
                try
                {
                    if (!receivedChannel2.Writer.TryWrite((message, descriptor)))
                    {
                        while (receivedChannel2.Writer.WaitToWriteAsync(cts.Token).AsTask().ConfigureAwait(false).GetAwaiter().GetResult())
                        {
                            if (receivedChannel2.Writer.TryWrite((message, descriptor)))
                            {
                                return;
                            }
                        }
                    }
                }
                catch (OperationCanceledException)
                {
                    ////Ignore
                }

                return;
            }

            try
            {
                if (!receivedChannel.Writer.TryWrite((message, descriptor)))
                {
                    while (receivedChannel.Writer.WaitToWriteAsync(cts.Token).AsTask().ConfigureAwait(false).GetAwaiter().GetResult())
                    {
                        if (receivedChannel.Writer.TryWrite((message, descriptor)))
                        {
                            return;
                        }
                    }
                }
            }
            catch (OperationCanceledException)
            {
                ////Ignore
            }
        }

        public void Dispose()
        {
            if (!cts.IsCancellationRequested)
            {
                cts.Cancel();
            }
        }

        private async Task Sending(CancellationToken cancellationToken)
        {
            try
            {
                while (await publishedChannel.Reader.WaitToReadAsync(cancellationToken))
                {
                    while (publishedChannel.Reader.TryRead(out var message))
                    {
                        try
                        {
                            var result = await sender.SendAsync(message);
                            if (!result.Succeeded)
                            {
                                logger.LogError(result.Exception, $"An exception occured while publishing a message, reason:{result.ToString()}. message id:{message.Origin.GetId()}");
                            }
                        }
                        catch (Exception ex)
                        {
                            logger.LogError(ex, $"An exception occurred when sending a message to the MQ. Id:{message.DbId}");
                        }
                    }
                }
            }
            catch (OperationCanceledException)
            {
                // expected
            }
        }

        private async Task Sending2(CancellationToken cancellationToken)
        {
            try
            {
                while (await publishedChannel2.Reader.WaitToReadAsync(cancellationToken))
                {
                    while (publishedChannel2.Reader.TryRead(out var message))
                    {
                        try
                        {
                            var result = await sender.SendAsync(message);
                            if (!result.Succeeded)
                            {
                                logger.LogError(result.Exception, $"An exception occured while publishing a message, reason:{result.ToString()}. message id:{message.Origin.GetId()}");
                            }
                        }
                        catch (Exception ex)
                        {
                            logger.LogError(ex, $"An exception occurred when sending a message to the MQ. Id:{message.DbId}");
                        }
                    }
                }
            }
            catch (OperationCanceledException)
            {
                // expected
            }
        }

        private async Task Processing(CancellationToken cancellationToken)
        {
            try
            {
                while (await receivedChannel.Reader.WaitToReadAsync(cancellationToken))
                {
                    while (receivedChannel.Reader.TryRead(out var message))
                    {
                        try
                        {
                            await executor.DispatchAsync(message.Item1, message.Item2, cancellationToken);
                        }
                        catch (OperationCanceledException)
                        {
                            ////expected
                        }
                        catch (Exception e)
                        {
                            logger.LogError(e, $"An exception occurred when invoke subscriber. MessageId:{message.Item1.DbId}");
                        }
                    }
                }
            }
            catch (OperationCanceledException)
            {
                // expected
            }
        }

        private async Task Processing2(CancellationToken cancellationToken)
        {
            try
            {
                while (await receivedChannel2.Reader.WaitToReadAsync(cancellationToken))
                {
                    while (receivedChannel2.Reader.TryRead(out var message))
                    {
                        try
                        {
                            await executor.DispatchAsync(message.Item1, message.Item2, cancellationToken);
                        }
                        catch (OperationCanceledException)
                        {
                            ////expected
                        }
                        catch (Exception e)
                        {
                            logger.LogError(e, $"An exception occurred when invoke subscriber. MessageId:{message.Item1.DbId}");
                        }
                    }
                }
            }
            catch (OperationCanceledException)
            {
                // expected
            }
        }
    }
}
